package thread.pool.core.executor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.Ordered;
import org.springframework.util.CollectionUtils;
import thread.pool.core.util.ResizableCapacityLinkedBlockIngQueue;
import thread.pool.core.util.ThrowableUtil;
import thread.pool.core.vo.ThreadPoolStatus;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * 动态线程池类
 * 这里实现了DisposableBean是为了在bean销毁时能够同时销毁掉线程池资源
 * 实现InitializingBean 是为了在bean初始化阶段属性注入之后去创建线程池资源
 * 过早地创建线程池 是为了防止其他bean创建完成的时候执行初始化任务依赖此线程池 保证了线程池在业务代码中正常使用
 */
public class DynamicThreadPoolExecutor implements DisposableBean, ApplicationContextAware, InitializingBean, Ordered {
    private static final Logger logger = LoggerFactory.getLogger(DynamicThreadPoolExecutor.class);

    private String taskDescribe;
    private Integer corePoolSize = 2;
    private Integer maxPoolSize = 20;
    private Long keepAliveTime = 10L;
    private TimeUnit unit;
    private Integer dequeSize;
    private String poolName;
    private Integer beatTime;
    private Integer alarmSize;
    private ThreadPoolExecutor threadPoolTaskExecutor;
    private ResizableCapacityLinkedBlockIngQueue<Runnable> blockIngQueue;
    private volatile boolean STOP;
    private final Object poolSizeMonitor;
    // 检测线程池状态频率 毫秒单位
    private volatile long checkFrequency = 1000;
    // 拒绝策略  默认由主线程执行 然后告警通知程序员队列异常
    private static RejectedExecutionHandler rejectedExecutionHandler =
            new ThreadPoolExecutor.CallerRunsPolicy();

    public DynamicThreadPoolExecutor() {
        this.unit = TimeUnit.SECONDS;
        this.dequeSize = 500;
        this.poolName = "pool";
        this.beatTime = 10;
        this.threadPoolTaskExecutor = null;
        this.blockIngQueue = null;
        this.STOP = false;
        this.alarmSize = 400;//告警阀值大于400 且默认时间内一直都是高于这个值 则认为应该关注下队列大小是否应该调整
        this.poolSizeMonitor = new Object();// 对象初始化时创建锁
    }


    /**
     * 资源要销毁
     */
    public void destroy() throws Exception {
        this.shutDown();
        logger.info(">>>>>>>>>线程池【{}】销毁成功", this.getPoolName());
    }

    public int getOrder() {
        return 0;
    }

    /**
     * bean成员属性构造注入之后
     */
    public void afterPropertiesSet() throws Exception {
        this.start();
    }


    /**
     * 线程池初始化
     * 主要做了两件事 1 初始化线程池 2 初始化监控线程
     */
    private void start() {
        logger.info("======= thread-pool {} start init>>>>>>>>>>>>>>>>>>>>>", this.poolName);
        this.blockIngQueue = new ResizableCapacityLinkedBlockIngQueue<>(this.dequeSize);

        ThreadPoolExecutor executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, this.keepAliveTime,
                this.unit, this.blockIngQueue,
                new DefaultThreadFactory(this.poolName),
                rejectedExecutionHandler);
        logger.info("======= thread-pool {} init success >>>>>>>>>>>>>>>>>>>>>", this.poolName);
        this.threadPoolTaskExecutor = executor;
        logger.info(this + "\n");

        // 新启线程负责检测线程池状态
        (new Thread(() -> {
            if (this.alarmSize == null) {
                this.alarmSize = this.dequeSize;
            }


            while (!this.STOP) {


                // 活跃线程数
                int activeCount = this.threadPoolTaskExecutor.getActiveCount();
                // 此时核心线程用尽 队列积压满了
                if (activeCount > getCorePoolSize()) {
                    invokeAlarmPostProcessor();
                }
                // 队列中任务数量
                int size = this.threadPoolTaskExecutor.getQueue().size();
                if (size >= this.alarmSize) {
                    logger.warn("【线程池队列告警】线程池状态：{}", this);
                    // 这里是预留接口给程序员使用 这里可以将线程信息传递给使用者 由使用者决定如何告警（短信 or 邮件？）
                    invokeAlarmPostProcessor();
                }

                try {
                    Thread.sleep(checkFrequency);
                } catch (InterruptedException e) {
                    logger.error("thread fail {}", ThrowableUtil.toString(e));
                    this.STOP = true;
                }
            }

        })).start();
    }

    /**
     * 执行告警处理器
     * 模仿spring框架 提供给程序员扩展点 自定义实现业务逻辑
     */
    private void invokeAlarmPostProcessor() {
        ThreadPoolStatus threadPoolStatus = this.convertThreadPoolStatus();

        // 获取所有的告警后置处理器bean
        List<AlarmPostProcessor> alarmPostProcessors = obtainAlarmPostProcessorList();
        if (alarmPostProcessors != null) {
            for (AlarmPostProcessor postProcessor : alarmPostProcessors) {
                postProcessor.postProcessor(threadPoolStatus);
            }
        }

    }

    public ThreadPoolExecutor getExecutor() {
        return this.threadPoolTaskExecutor;
    }

    /**
     * 刷新线程池状态
     */
    public void refresh() {
        synchronized (this.poolSizeMonitor) {
            logger.info(">>>>>>>>>>>> 开始刷新线程池参数 >>>>>>>>>>>");
            if (!this.threadPoolTaskExecutor.isShutdown() && !this.threadPoolTaskExecutor.isTerminated()) {
                this.threadPoolTaskExecutor.setCorePoolSize(this.corePoolSize);
                this.threadPoolTaskExecutor.setMaximumPoolSize(this.maxPoolSize);
                this.updateDequeSize(this.dequeSize);
                logger.info(">>>>>>>>>>>> 线程池参数刷新完毕 >>>>>>>>>>>");
                logger.info(this + "\n");
                this.STOP = false;
            } else {
                throw new RuntimeException("线程池状态异常 不可刷新");
            }
        }
    }

    protected void updateDequeSize(Integer dequeSize) {
        if (this.blockIngQueue == null) {
            throw new RuntimeException(" blockIngQueue is null ");
        } else {
            this.blockIngQueue.setCapacity(dequeSize);
        }
    }

    public void execute(Runnable runnable) {
        ThreadPoolExecutor threadPoolExecutor = this.getExecutor();
        logger.info(">>>>>>>>> poolName:{},activeCount：{},queueTaskCount：{}", this.getPoolName(), threadPoolExecutor.getActiveCount(), threadPoolExecutor.getQueue().size());
        threadPoolExecutor.execute(runnable);
    }

    public <E> Future<E> submit(Callable<E> callable) {
        ThreadPoolExecutor executor = this.getExecutor();
        return executor.submit(callable);
    }

    protected void shutDown() {
        synchronized (this.poolSizeMonitor) {
            ThreadPoolExecutor executor = this.getExecutor();
            executor.shutdown();
            if (executor.isShutdown()) {
                executor = null;
            }

        }
    }

    public RejectedExecutionHandler getRejectedExecutionHandler() {
        return rejectedExecutionHandler;
    }

    public void setRejectedExecutionHandler(RejectedExecutionHandler rejectedExecutionHandler) {
        DynamicThreadPoolExecutor.rejectedExecutionHandler = rejectedExecutionHandler;
    }

    public long getCheckFrequency() {
        return checkFrequency;
    }

    public void setCheckFrequency(long checkFrequency) {
        this.checkFrequency = checkFrequency;
    }

    public Integer getBeatTime() {
        return this.beatTime;
    }

    public void setBeatTime(Integer beatTime) {
        this.beatTime = beatTime;
    }

    public Integer getCorePoolSize() {
        return this.corePoolSize;
    }

    public void setCorePoolSize(Integer corePoolSize) {
        this.corePoolSize = corePoolSize;
    }

    public Integer getMaxPoolSize() {
        return this.maxPoolSize;
    }

    public void setMaxPoolSize(Integer maxPoolSize) {
        this.maxPoolSize = maxPoolSize;
    }

    public Integer getDequeSize() {
        return this.dequeSize;
    }

    public void setDequeSize(Integer dequeSize) {
        this.dequeSize = dequeSize;
    }

    public String getPoolName() {
        return this.poolName;
    }

    public void setPoolName(String poolName) {
        this.poolName = poolName;
    }

    public Long getKeepAliveTime() {
        return this.keepAliveTime;
    }

    public void setKeepAliveTime(Long keepAliveTime) {
        this.keepAliveTime = keepAliveTime;
    }

    public TimeUnit getUnit() {
        return this.unit;
    }

    public void setUnit(TimeUnit unit) {
        this.unit = unit;
    }

    public Integer getAlarmSize() {
        return this.alarmSize;
    }

    public void setAlarmSize(Integer alarmSize) {
        this.alarmSize = alarmSize;
    }

    public String getTaskDescribe() {
        return this.taskDescribe;
    }

    public void setTaskDescribe(String taskDescribe) {
        this.taskDescribe = taskDescribe;
    }

    public String toString() {
        return "线程池信息{ poolName='" + this.poolName + ", corePoolSize=" + this.corePoolSize + ", maxPoolSize=" + this.maxPoolSize + ", keepAliveTime=" + this.keepAliveTime + ", unit=" + this.unit + ", dequeSize=" + this.dequeSize + ",排队任务数=" + this.threadPoolTaskExecutor.getQueue().size() + ",活跃线程数=" + this.threadPoolTaskExecutor.getActiveCount() + '}';
    }

    private static ApplicationContext applicationContext;

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        DynamicThreadPoolExecutor.applicationContext = applicationContext;
    }

    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    public static <T> Map<String, T> getBeansOfType(Class<T> tClass) {
        return applicationContext.getBeansOfType(tClass);
    }

    /**
     * 转换线程状态信息
     *
     * @return 状态信息
     */
    private ThreadPoolStatus convertThreadPoolStatus() {
        ThreadPoolStatus threadPoolStatus = new ThreadPoolStatus();

        threadPoolStatus.setCorePoolSize(this.getCorePoolSize());
        threadPoolStatus.setPoolName(this.getPoolName());
        threadPoolStatus.setActiveCount(this.threadPoolTaskExecutor.getActiveCount());
        threadPoolStatus.setQueueCount(this.threadPoolTaskExecutor.getQueue().size());
        threadPoolStatus.setMaxPoolSize(this.getMaxPoolSize());
        threadPoolStatus.setAlarmSize(this.getAlarmSize());
        threadPoolStatus.setTaskDescribe(this.getTaskDescribe());
        threadPoolStatus.setDequeSize(this.getDequeSize());

        return threadPoolStatus;
    }

    /**
     * 获取spring中所有实现了AlarmPostProcessor接口的bean
     *
     * @return bean集合
     */
    public static List<AlarmPostProcessor> obtainAlarmPostProcessorList() {
        Map<String, AlarmPostProcessor> beansOfType = getBeansOfType(AlarmPostProcessor.class);
        if (CollectionUtils.isEmpty(beansOfType)) {
            return null;
        }
        return new ArrayList<>(beansOfType.values());
    }

    /**
     * 自定义线程工厂
     */
    static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        DefaultThreadFactory(String poolName) {
            SecurityManager s = System.getSecurityManager();
            this.group = s != null ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            this.namePrefix = poolName + "-" + poolNumber.getAndIncrement() + "-thread-";
        }

        public Thread newThread(Runnable r) {
            return new Thread(this.group, r, this.namePrefix + this.threadNumber.getAndIncrement(), 0L);
        }
    }
}
